package cn.doitedu.api;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/27
 * @Desc: 学大数据，上多易教育
 *   观察：
 *      事件时间
 *      watermark
 *      及事件时间语义的定时器
 **/
public class _17_EventTime_Watermark_Timer_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/doit46_ckpt");

        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 7788);
        // a,1000
        DataStream<Tuple2<String, Long>> tuples = stream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] split = value.split(",");

                return Tuple2.of(split[0], Long.parseLong(split[1]));
            }
        });

        // 添加了一个算子：用于抽取数据中的时间字段，并生成watermark
        DataStream<Tuple2<String, Long>> watermarked = tuples.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(2))  // 允许2秒乱序： watermark = data.ts - 2s
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                return element.f1;
                            }
                        })
        );


        watermarked.keyBy(tp->tp.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> value, KeyedProcessFunction<String, Tuple2<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {

                        long currentWatermark = ctx.timerService().currentWatermark();  // 当前的watermark
                        System.out.println("当前 watermark: " + currentWatermark);

                        Long ctxTimestamp = ctx.timestamp();  // 当前收到的数据本身的时间戳
                        System.out.println("当前 ctxTimestamp: " + ctxTimestamp);

                        long currentProcessingTime = ctx.timerService().currentProcessingTime();  // 当前的处理时间 = 系统的时间
                        System.out.println("当前 currentProcessingTime," + currentProcessingTime);
                        System.out.println("当前系统时间： " + System.currentTimeMillis());

                        if("x".equals(value.f0)){
                            long timerTime = currentWatermark + 5000;
                            System.out.println("注册了一个定时器,时间为： " +  timerTime );
                            ctx.timerService().registerEventTimeTimer(timerTime);
                        }

                        out.collect(value.f0);
                    }


                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, Tuple2<String, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("定时器被触发了");
                        System.out.println("onTimer 方法中传入的 timestamp: " + timestamp);  // 定时器被触发时的时间（或event time,或processing time)

                        Long ctxTimestamp = ctx.timestamp();
                        System.out.println("onTimer 中的 ctxTimestamp: " + ctxTimestamp);  //

                        long wt = ctx.timerService().currentWatermark();  // 当前的watermark
                        System.out.println("onTimer 中的 watermark: " + wt);

                        long pt = ctx.timerService().currentProcessingTime();   // 当前的processing time
                        System.out.println("onTimer 中的 ProcessingTime: " + pt);


                    }
                });

        env.execute();

    }
}
